package rm3;


import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.PriorityQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import server.ServerImpl;

public class ReliableUDPListener implements Runnable{
	DatagramSocket listenerSocket = null;
	byte[] requested, ack;
	public static ServerImpl montreal;
	public static ServerImpl vancouver;
	public static ServerImpl toronto;
	ExecutorService serverthreadPool;
	static ArrayList<Integer> holdBackIndex = new ArrayList<Integer>();
	static PriorityQueue<Message> holdBack = new PriorityQueue<Message>();
	long [] arrayDuplicated;
	int ArrayIndex = 0;
	long sequencer;
	public static ConcurrentHashMap<Long, String> storedReply = new ConcurrentHashMap <Long, String> ();



	public ReliableUDPListener(DatagramSocket socket, ServerImpl montrealObj, ServerImpl torontoObj, ServerImpl vancouverObj){
		listenerSocket = socket;
		montreal = montrealObj;
		toronto = torontoObj;
		vancouver = vancouverObj;
		serverthreadPool = Executors.newFixedThreadPool(20); 
		arrayDuplicated = new long[1000] ;
		sequencer = 1;
	}

	public void run() {
		try{

			while(true){
				requested = new byte[64];
				ack = new byte[16];
				// receive client's request
				DatagramPacket clientRequest = new DatagramPacket(requested, requested.length);
				listenerSocket.receive(clientRequest); 


				Message m = new Message(0);
				m.parseMessage(clientRequest.getData());
				System.out.println("service requested at RM3: "+  (new String (clientRequest.getData(),0, clientRequest.getLength())));

				if(m.getRequestType().equals("reply"))
					serverthreadPool.execute( new Reply(m.getSequencerID(), clientRequest, listenerSocket));

				else {
					//send ack
					ack =  Integer.toString((int) m.getSequencerID()).getBytes();

					DatagramPacket replytoClient = new DatagramPacket(ack, ack.length,
							clientRequest.getAddress(), clientRequest.getPort());

					listenerSocket.send(replytoClient);
					// send request to holdback queue
					try {
						replyAndReceiveAck(clientRequest);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (ExecutionException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}

				}
			}

		}catch(SocketException e){
			System.out.println("Server Socket Exception: " + e.getMessage());
		}catch (IOException e) {
			System.out.println("IO: " + e.getMessage());
		}finally {
			if(listenerSocket != null) listenerSocket.close();
		}
		System.out.println("Exiting server's try catch!");

	}

	
	public void replyAndReceiveAck(DatagramPacket clientRequest) throws InterruptedException, ExecutionException, SocketException, UnknownHostException{

		Message m = new Message(0);
		m.parseMessage(clientRequest.getData());
		boolean flag = false;
		String answer = "-99";
		if(!IsDuplicated(m))
		{
			if(m.getSequencerID() == sequencer)
			{
				System.out.println("--------------------------> message :" + m.getSequencerID());
				answer = sendRequesetToServer(m);
				storedReply.put(sequencer, answer);
				System.out.println("Answer after queue for "+  m.getSequencerID() +" : "+answer);
				arrayDuplicated[ArrayIndex] = m.getSequencerID();
				ArrayIndex++;
				sequencer++;
				flag = true;

			}
			else if(m.getSequencerID() > sequencer)
			{
				holdBack.add(m);
				arrayDuplicated[ArrayIndex] = m.getSequencerID();
				ArrayIndex++;

			}
			while(flag && (!holdBack.isEmpty()))
			{

				System.out.println("************************> message :" + holdBack.peek().getSequencerID());
				if(holdBack.peek().getSequencerID() == sequencer)
				{
					System.out.println("--------------------------> message :" + holdBack.peek().getSequencerID());
					answer = sendRequesetToServer(holdBack.peek());
					storedReply.put(sequencer, answer);
					System.out.println("Answer after queue for "+  holdBack.peek().getSequencerID()+" : "+answer);
					holdBack.poll();
					sequencer++;
				}
				else
				{
					flag = false;
				}

			}
		}
		else 
		{
			System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>> this is Duplicated " + m.getSequencerID());

		}

	}

	public String sendRequesetToServer(Message message){
		String result = "";
		char store;
		store = message.getCustomerID().charAt(0);
		if(store == 'T') {
			switch(message.getRequestType()){
			case "buy":
				result = ReliableUDPListener.toronto.buy(message.getCustomerID(), message.getItemID(),
						message.getNumberOfItems());
				break;
			case "checkStock":
				result = ReliableUDPListener.toronto.checkStock(message.getItemID());
				break;
			case "return":
				result = ReliableUDPListener.toronto.returnStock(message.getCustomerID(), message.getItemID(),
						message.getNumberOfItems());
				break;
			case "exchange":
				result = ReliableUDPListener.toronto.exchange(message.getCustomerID(),
						message.getBoughtItemID(),message.getBoughtNumber(),
						message.getItemID(),message.getNumberOfItems());
				break;
			}
		}else if(store == 'M'){
			switch(message.getRequestType()){
			case "buy":
				result = ReliableUDPListener.montreal.buy(message.getCustomerID(), message.getItemID(),
						message.getNumberOfItems());
				break;
			case "checkStock":
				result = ReliableUDPListener.montreal.checkStock(message.getItemID());
				break;
			case "return":
				result = ReliableUDPListener.montreal.returnStock(message.getCustomerID(), message.getItemID(),
						message.getNumberOfItems());
				break;
			case "exchange":
				result = ReliableUDPListener.montreal.exchange(message.getCustomerID(),
						message.getBoughtItemID(),message.getBoughtNumber(),
						message.getItemID(),message.getNumberOfItems());
				break;
			}
		}else if(store == 'V'){
			switch(message.getRequestType()){
			case "buy":
				result = ReliableUDPListener.vancouver.buy(message.getCustomerID(), message.getItemID(),
						message.getNumberOfItems());
				break;
			case "checkStock":
				result = ReliableUDPListener.vancouver.checkStock(message.getItemID());
				break;
			case "return":
				result = ReliableUDPListener.vancouver.returnStock(message.getCustomerID(), message.getItemID(),
						message.getNumberOfItems());
				break;
			case "exchange":
				result = ReliableUDPListener.vancouver.exchange(message.getCustomerID(),
						message.getBoughtItemID(),message.getBoughtNumber(),
						message.getItemID(),message.getNumberOfItems());
				break;
			}

		}
		System.out.println("Received from replica: "+ result);
		return result;
	}

	public boolean IsDuplicated(Message message)
	{
		Message localmessage = new Message(0);
		localmessage = message;
		boolean Duplicated = false;
		for(int i = 0; i < arrayDuplicated.length; i++) 
		{  

			if(arrayDuplicated[i] == localmessage.getSequencerID())
			{
				Duplicated = true;
			}
		}  
		return Duplicated;
	}

	class Reply implements Runnable{
		byte[] replyFromreplica = new byte[16];
		DatagramPacket tosend;
		DatagramPacket client;
		DatagramSocket serverSocket;
		long id;
		Reply(long sequenceNumber, DatagramPacket fromclient, DatagramSocket server){
			client = fromclient;
			id = sequenceNumber;
			serverSocket = server;
		}
		@Override
		public 	void run(){
			if(sequencer>id){
				System.out.println("Retrieving reply from stored reply.");
				replyFromreplica = storedReply.get(id).getBytes();
				tosend = new DatagramPacket(replyFromreplica, replyFromreplica.length,
						client.getAddress(), client.getPort());

			} else {
				System.out.println("Making FE rerequest for a reply.");
				try {
					Thread.sleep(500);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				replyFromreplica ="999999".getBytes();
				tosend = new DatagramPacket(replyFromreplica, replyFromreplica.length,
						client.getAddress(), client.getPort());
			}
			try {
				serverSocket.send(tosend);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}


}

